package com.chinamobile.websocket.websocketdispatcher.config;

import com.chinamobile.websocket.domain.bo.WebsocketKafkaMessageBo;
import com.chinamobile.websocket.domain.dto.MessageDto;
import com.chinamobile.websocket.websocketdispatcher.common.MessageDistributorConstants;
import com.chinamobile.websocket.websocketdispatcher.distributor.MessageDistributor;
import com.chinamobile.websocket.websocketdispatcher.distributor.kafka.KafkaMessageDistributor;
import com.chinamobile.websocket.websocketdispatcher.distributor.kafka.WebsocketListening;
import lombok.RequiredArgsConstructor;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.KafkaTemplate;

/**
 * @className: KafkaMessageDistributorConfig
 * @description 基于 kafka 的消息分发器
 * @author: fanzhijun
 * @date: 2024/1/22
 **/

@ConditionalOnClass(KafkaTemplate.class)
@ConditionalOnProperty(prefix = MessageDistributorConstants.DISTRIBUTOR_PREFIX, name = "message-distributor",
        havingValue = MessageDistributorConstants.KAFKA)
@Configuration(proxyBeanMethods = false)
@RequiredArgsConstructor
public class KafkaMessageDistributorConfig {
    @Bean
    @ConditionalOnMissingBean(MessageDistributor.class)
    public KafkaMessageDistributor<MessageDto> messageDistributor(KafkaTemplate<String, WebsocketKafkaMessageBo> kafkaTemplate) {
        return new KafkaMessageDistributor<>(kafkaTemplate);
    }


    @Bean
    @ConditionalOnClass(MessageDistributor.class)
    public WebsocketListening websocketListening(MessageDistributor<MessageDto> messageDistributor) {
        WebsocketListening listening = new WebsocketListening();
        listening.setMessageDistributor(messageDistributor);
        return listening;
    }


}
